package net.neptech.service.impl;

import jakarta.annotation.Resource;
import lombok.extern.slf4j.Slf4j;

import net.neptech.model.appointment.Appointment;
import net.neptech.dto.order.DelayMessage;
import net.neptech.service.DelayQueueService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.data.redis.core.HashOperations;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.SendResult;
import org.springframework.stereotype.Service;

import jakarta.annotation.PostConstruct;
import jakarta.annotation.PreDestroy;

import java.time.LocalDateTime;
import java.util.Map;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

/**
 * 延迟队列服务
 * 实现延迟消息的发送和管理功能
 *
 * 实现原理：
 * 1. 使用内存存储延迟消息（生产环境建议使用Redis）
 * 2. 定时任务扫描到期的延迟消息
 * 3. 将到期消息发送到Kafka进行处理
 *
 * @author neptech
 */



@Slf4j
@Service
public class DelayQueueServiceImpl implements DelayQueueService {

    @Autowired
    private KafkaTemplate<String, DelayMessage> kafkaTemplate;

    @Value("${delayed.queue.topic}")
    private String delayTopic;

    @Value("${delayed.queue.max-retry-count}")
    private Integer maxRetryCount;

    @Value("${delayed.queue.check-interval-seconds}")
    private Integer checkIntervalSeconds;

    @Resource
    private RedisTemplate<String,DelayMessage> delayMessageRedisTemplate;
    /**
     * 定时任务执行器
     */
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(2);

    /**
     * 初始化延迟队列服务
     * 启动定时任务扫描到期消息
     */
    @PostConstruct
    public void init() {
        log.info("初始化延迟队列服务，检查间隔: {}秒", checkIntervalSeconds);
        // 启动定时任务，定期检查到期的延迟消息
        scheduler.scheduleWithFixedDelay(
                this::processExpiredMessages,
                checkIntervalSeconds,
                checkIntervalSeconds,
                TimeUnit.SECONDS
        );
    }

    /**
     * 发送延迟消息
     *
     * @param messageType 消息类型
     * @param content 消息内容
     * @param delayMinutes 延迟时间（分钟）
     * @return 消息ID
     */
    @Override
    public String sendDelayMessage(String messageType, Appointment content, int delayMinutes) {
        return sendDelayMessage(messageType, content, delayMinutes, null);
    }

    /**
     * 发送延迟消息
     *
     * @param messageType 消息类型
     * @param content 消息内容
     * @param delayMinutes 延迟时间（分钟）
）
     * @return 消息ID
     */
    @Override
    public String sendDelayMessage(String messageType, Appointment content, int delayMinutes, Long orderId) {

            String messageId = content.getOrderId().toString();
            // 拉长延迟时间 处理为30s
            LocalDateTime executeTime = LocalDateTime.now().plusMinutes(delayMinutes).plusSeconds(30);

            DelayMessage delayMessage = DelayMessage.builder()
                    .messageId(messageId)
                    .messageType(messageType)
                    .content(content) // 传预约单吧
                    .executeTime(executeTime)
                    .createTime(LocalDateTime.now()) // 记录预约单时间
                    .retryCount(0)
                    .maxRetryCount(maxRetryCount)
                    .status("PENDING")
                    .build();
            //放入redis中
            delayMessageRedisTemplate.opsForHash().put("delayed_messages",messageId,delayMessage);
            log.info("发送延迟消息成功，messageId: {}, messageType: {}, executeTime: {}",
                    messageId, messageType, executeTime);
            return messageId;


    }

    /**
     * 取消延迟消息
     *
     * @param messageId 消息ID
     * @return 是否取消成功
     */
//    public boolean cancelDelayMessage(String messageId) {
//        if (messageId == null || messageId.trim().isEmpty()) {
//            return false;
//        }
//        DelayMessage removed = delayMessages.remove(messageId);
//        boolean success = removed != null;
//        log.info("取消延迟消息，messageId: {}, 结果: {}", messageId, success ? "成功" : "失败");
//        return success;
//    }


    /**
     * 处理到期的延迟消息
     * 定时任务调用此方法扫描并处理到期的消息
     */
    private void processExpiredMessages() {
        LocalDateTime now = LocalDateTime.now();
        HashOperations<String, String, DelayMessage> delayedOptions = delayMessageRedisTemplate.opsForHash();
        Map<String, DelayMessage> delayedMessages = delayedOptions.entries("delayed_messages");
        System.out.println(delayedMessages);
        delayedMessages.values().stream()
                .filter(message -> message.isTimeToExecute())
                .forEach(message -> {
                    try {
                        // 发送消息到Kafka
                        CompletableFuture<SendResult<String, DelayMessage>> future =
                                kafkaTemplate.send(delayTopic, message.getMessageId(), message);
                        future.whenComplete((result, ex) -> {
                            if (ex == null) {
                                // 发送成功，从内存中移除
                                delayMessageRedisTemplate.opsForHash().delete("delayed_messages", message.getMessageId().toString());
                                log.info("延迟消息发送成功，messageId: {}, messageType: {}",
                                        message.getMessageId(), message.getMessageType());
                            } else {
                                // 发送失败，处理重试逻辑
                                handleSendFailure(message, ex);
                            }
                        });

                    } catch (Exception e) {
                        log.error("处理延迟消息异常，messageId: {}", message.getMessageId(), e);
                        handleSendFailure(message, e);
                    }
                });
    }

    /**
     * 处理消息发送失败的情况
     *
     * @param message 延迟消息
     * @param throwable 异常信息
     */
    private void handleSendFailure(DelayMessage message, Throwable throwable) {
        message.incrementRetryCount();

        if (message.canRetry()) {
            // 可以重试，延迟1分钟后重试
            message.setExecuteTime(LocalDateTime.now().plusMinutes(1));
            message.setStatus("RETRY");
            log.warn("延迟消息发送失败，将重试，messageId: {}, retryCount: {}",
                    message.getMessageId(), message.getRetryCount(), throwable);
        } else {
            // 超过最大重试次数，标记为失败并移除
            delayMessageRedisTemplate.opsForHash().delete("delayed_messages", message.getMessageId());
            message.setStatus("FAILED");
            log.error("延迟消息发送失败，超过最大重试次数，messageId: {}, retryCount: {}",
                    message.getMessageId(), message.getRetryCount(), throwable);
        }
    }

    @PreDestroy
    public void destroy() {
        scheduler.shutdown();
        log.info("延迟队列服务已关闭");
    }
}
